package mq

import (
	"github.com/Shopify/sarama"
	"goTool/common"
)

type KafkaProducer struct {
	ProductConn sarama.SyncProducer
}

func InitProducer(cfg ProducerCfg) *KafkaProducer {
	producerCfg := sarama.NewConfig()
	producerCfg.Producer.RequiredAcks = sarama.WaitForAll          // 等待从Kafka消息同步成功后,返回发送成功
	producerCfg.Producer.Partitioner = sarama.NewRandomPartitioner // 分区策略(这条消息要写入哪个分区)
	producerCfg.Producer.Return.Successes = true

	conn, err := sarama.NewSyncProducer(cfg.Address, producerCfg)
	common.IsError(err)

	return &KafkaProducer{ProductConn: conn}
}

/*
	topic: 主题
	data:  消息
	partition: 消息所属分区
	offset: 消息偏移量位置
*/
func (product *KafkaProducer) SendMessage(topic string, data []byte) (partition int32, offset int64, err error) {
	var msg sarama.ProducerMessage
	msg.Topic = topic
	msg.Value = sarama.StringEncoder(data)

	return product.ProductConn.SendMessage(&msg)
}
